package cn.doitedu.day05

import cn.doitedu.day01.utils.SparkUtil
import cn.doitedu.day03.beans.Student
import org.apache.spark.rdd.RDD

/**
 * @Date 22.4.1
 * @Created by HANGGE
 * @Description
 */
object C06_转换算子_GroupBy02 {
  def main(args: Array[String]): Unit = {
    // 统计每个班级的平均分
    val sc = SparkUtil.getSc
    val rdd1 = sc.textFile("data/students.csv")

    val rdd2 = rdd1.map(line => {
      val arr = line.split(",")
      Student(arr(0).toInt, arr(1), arr(2).toInt, arr(3), arr(4).toDouble, arr(5))
    })
    // 统计每个班级的平均分
    // 1) 按照班级分组
    /**
     * tp._1  组名
     * tp._2  组内的多条数据   迭代器
     */
    val rdd3: RDD[(String, Iterable[Student])] = rdd2.groupBy(_.grade)
    rdd2.groupBy(_.age>20)
    rdd2.groupBy(stu=>(stu.age>20 && stu.score>88))
    rdd2.groupBy(_.age%3)

    // 算子中的函数被封装到Task中  , Task负责处理属于自己的数据  运行在不同的Worker上
    // 每个Task负责处理的数据是有限的 , 不会太大
    // 本地集合存储所有的数据
    val rdd4 = rdd3.map(tp=>{
      val grade = tp._1  // 班级
      var len = 0
      var sum =0d
      //将班级下所有的分数转换成List集合
     tp._2.foreach(stu=>{
       // 累加计算数据
       sum+=stu.score
       len+=1
     })
      // 求平均成绩
      val avg:Double = sum/len
      // 返回结果
      (grade , avg)
    })
    // 执行行动算子  触发执行
    rdd4.foreach(println)


  }

}
